package com.sun.clients.config;

import com.sun.clients.properties.KafkaConsumerProperty;
import lombok.RequiredArgsConstructor;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.Properties;

@Configuration
@RequiredArgsConstructor(onConstructor = @__(@Autowired))
public class KafkaConsumerConfig {

    private final KafkaConsumerProperty kafkaConsumerProperty;

    @Bean("kafkaConsumer")
    public KafkaConsumer<String, String> kafkaConsumer() {
        Properties props = new Properties();
        // 服务地址，多个以隔开（必填） 示例：kafka1:9092,kafka2:9092,kafka3:9092
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaConsumerProperty.getBootstrapServers());
        // 消费者组（必填） 示例：test-group 默认：空字符串
        props.put(ConsumerConfig.GROUP_ID_CONFIG, kafkaConsumerProperty.getGroupId());
        // key反序列化方式（必填）
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        // value反序列化方式（必填）
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);

        // 是否自动提交Offset 默认：true
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, kafkaConsumerProperty.getEnableAutoCommit());
        // 自动提交间隔时间 默认：5000 单位：毫秒
        props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, kafkaConsumerProperty.getAutoCommitInterval());
        // 最大拉取数量 默认：500
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, kafkaConsumerProperty.getMaxPollRecords());
        // 最大拉取超时时间 默认：300000 单位：毫秒
        props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, kafkaConsumerProperty.getMaxPollInterval());
        // 连接超时时间 默认：45000 单位：毫秒
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, kafkaConsumerProperty.getSessionTimeout());
        // 等待请求响应的最长时间 默认：300000 单位：毫秒
        props.put(ConsumerConfig.REQUEST_TIMEOUT_MS_CONFIG, kafkaConsumerProperty.getRequestTimeout());
        // 发送心跳请求间隔 默认：3000 单位：毫秒
        props.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, kafkaConsumerProperty.getHeartbeatInterval());

        // 偏移量读取位置 默认：latest
        // latest：当各分区下存在已提交的 offset 时，从提交的 offset 开始消费；无提交的 offset 时，消费该分区下新产生的数据（默认值）
        // earliest：当各分区下存在已提交的 offset 时，从提交的 offset 开始消费；无提交的 offset 时，从头开始消费
        // none：当各分区都存在已提交的 offset 时，从 offset 后开始消费；只要有一个分区不存在已提交的offset，则直接抛出NoOffsetForPartitionException异常
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, kafkaConsumerProperty.getAutoOffsetReset());
        return new KafkaConsumer<>(props);
    }

}
